package com.galeno.demo;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/9/1121:53
 */
public class ConsumerOffsetTest2 {
    public static void main(String[] args) {
        String serverUrl = "galeno01:9092,galeno02:9092,galeno03:9092";
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, serverUrl);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //设置为手动提交
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        //如果没找到偏移量,从此参数指定位置开始消费
        properties.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "earliest");
        //消费者唯一的标识
        properties.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "f01");
        //消费者所属的组id
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "g01");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        //KafkaConsumer<String, String> consumer2 = new KafkaConsumer<String, String>(properties);

        //由消费者固定住要消费的topi以及分区
        TopicPartition first0 = new TopicPartition("first", 0);
        consumer.assign(Collections.singletonList(first0));
     //   consumer2.subscribe(Arrays.asList("first"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            if (!records.isEmpty()) {
                for (ConsumerRecord<String, String> record : records) {
                    //数据的逻辑处理
                    System.out.println("分区号:" + record.partition()
                            + "偏移量" + record.offset()
                            + "消息信息" + record.value()
                    );

                    // 如果在处理成功后，手动更新偏移量，则能实现  at least once 的数据处理语义
                    // 提交方式：同步调用
                    Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(first0, new OffsetAndMetadata(record.offset() + 1));
                    //consumer.commitSync(offsets);

                    // 提交方式：异步调用
                    consumer.commitAsync(offsets, new OffsetCommitCallback() {
                        @Override
                        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                            System.out.println("偏移量已经提交成功.....");
                        }
                    });
                }


            }


        }





    }
}
